package org.alex.mqtt.common.protocol.impl;

import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.mqtt.MqttConnAckMessage;
import io.netty.handler.codec.mqtt.MqttConnectMessage;
import io.netty.handler.codec.mqtt.MqttConnectReturnCode;
import io.netty.handler.codec.mqtt.MqttMessage;
import org.alex.mqtt.common.log.LogUtil;
import org.alex.mqtt.common.netty.MqttNettyUtils;
import org.alex.mqtt.common.remote.RemotingHelper;
import org.alex.mqtt.common.support.log.AlexLogger;
import org.alex.mqtt.common.protocol.MqttProcessor;
import org.alex.mqtt.common.MQTTConnection;
import org.alex.mqtt.common.utils.MqttMessageUtil;
import org.slf4j.Logger;

/**
 * mqtt 客户端连接逻辑处理
 * 强制约束：在未返回conAck之前，不接收其它任何mqtt协议报文（mqtt协议可以允许）
 * @author : alex
 */
public class ConnectProcessor implements MqttProcessor {

    private static final Logger log = AlexLogger.mqttLog;

    /**
     * mqtt 连接请求
     *
     */
    @Override
    public void processRequest(ChannelHandlerContext ctx, MqttMessage mqttMessage) {
        MqttConnectMessage connectMessage = (MqttConnectMessage) mqttMessage;

        MqttConnectReturnCode returnCode = null;

        int mqttVersion = connectMessage.variableHeader().version();
        String clientId = connectMessage.payload().clientIdentifier();
        String userName = connectMessage.payload().userName();
        byte[] password = connectMessage.payload().passwordInBytes();
        boolean cleanSession = connectMessage.variableHeader().isCleanSession();

        MQTTConnection mqttConnection = MqttNettyUtils.mqttConnection(ctx.channel());
        boolean sessionPresent = false;

        try {
            if (!versionValid(mqttVersion)) { // mqtt版本校验
                returnCode =  MqttConnectReturnCode.CONNECTION_REFUSED_UNACCEPTABLE_PROTOCOL_VERSION;
            } else if (!mqttConnection.clientIdVerify(clientId)) { // 客户端id校验
                returnCode = MqttConnectReturnCode.CONNECTION_REFUSED_IDENTIFIER_REJECTED;
            } else if (mqttConnection.onBlackList(RemotingHelper.getRemoteAddr(ctx.channel()), clientId)) { // 黑名单校验
                returnCode = MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED;
            } else if (!mqttConnection.login(clientId, userName, password)) { // 用户名密码校验
                returnCode = MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD;
            } else { // 所有检验通过
                // 1 设置心跳
                int heartbeatSec = connectMessage.variableHeader().keepAliveTimeSeconds();
                if (!mqttConnection.keepAlive(heartbeatSec)) {
                    LogUtil.warn(log,"[MQTT: CONNECT] -> 设置心跳失败,clientId:{},heartbeatSec:{}", clientId, heartbeatSec);
                    throw new Exception("设置心跳失败");
                }

                // 2 处理连接
                sessionPresent = mqttConnection.createOrReopenSession(connectMessage);
            }
        } catch (Exception e) {
            LogUtil.warn(log, "[MQTT: CONNECT] -> 服务不可用,原因 = {}", e);
            returnCode = MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE;
            MqttConnAckMessage ackMessage = MqttMessageUtil.getConnectAckMessage(returnCode, sessionPresent);
            ctx.writeAndFlush(ackMessage);
            ctx.close();
        }
    }

    /**
     * mqtt版本验证
     */
    private boolean versionValid(int mqttVersion) {
        if (mqttVersion == 3 || mqttVersion == 4) {
            return true;
        }
        return false;
    }
}
